package com.atguigu.flume.interceptor;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class LogTypeInterceptor implements Interceptor{


    //初始化
    @Override
    public void initialize() {

    }

    //对单一的event 进行处理
    @Override
    public Event intercept(Event event) {
        // head :body
        byte[] body = event.getBody();
        String info = new String(body , Charset.forName("UTF-8"));

        //获取数据
        // 获取头
        Map<String, String> headers = event.getHeaders();

        //根据数据的不同往不同的kafaka topic sink 数据
        if(info.contains("start")){
            headers.put("topic" ,"topic_start") ;
        }
        else {
            headers.put("topic" ,"topic_event") ;
        }

        return event;
    }

    //批量处理event
    @Override
    public List<Event> intercept(List<Event> events) {
        // 初始化输出
        ArrayList<Event> list = new ArrayList<Event>() ;
        for ( Event event:
              events) {
            list.add( event) ;
        }

        return list;
    }

    @Override
    public void close() {

    }
    //建立一个单列的对象
    public static class  Build implements  Interceptor.Builder  {

        @Override
        public Interceptor build() {
            return  new LogTypeInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}
